package cn.doitedu.rtdw.data_etl;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/8
 * @Desc: 学大数据，到多易教育
 * 视频播放行为分析etl任务
 * 任务的核心要点：
 * 1. 为视频播放相关事件数据 关联 视频信息维表
 * 2. 为视频播放相关事件数据增加： start_time字段
 * 3. 进行窗口聚合，求end_time，并依此降低写入doris的数据量和频率
 **/
public class Etl04_VideoPlayActionAnalyse_API {
    public static void main(String[] args) throws Exception {

        // 编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，来映射kafka中的dwd层的用户行为公共维度宽表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                           "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_province STRING,                     "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     pt AS proctime() ,                            "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 创建hbase连接器表，映射hbase中的视频信息维表
        tenv.executeSql("CREATE TABLE video_hbase( " +
                " video_id STRING, " +
                " f ROW<video_name STRING, video_type STRING, video_album STRING,video_author STRING, video_timelong BIGINT,create_time timestamp(3),update_time timestamp(3) >, " +
                " PRIMARY KEY (video_id) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_video_info',     " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");


        // 编写一个lookup join的sql语句，进行维度关联
        tenv.executeSql(
                " CREATE TEMPORARY VIEW joined_view AS              "
                        + " WITH tmp AS (                                     "
                        + " SELECT                                            "
                        + "   user_id,                                        "
                        + "   release_channel,                                "
                        + "   device_type,                                    "
                        + "   event_id,                                       "
                        + "   properties['play_id']  as play_id,              "
                        + "   properties['video_id'] as video_id,             "
                        + "   event_time,                                     "
                        + "   pt,                                             "
                        + "   rt                                              "
                        + " FROM dwd_kafka                                    "
                        + " WHERE event_id in ('video_play','video_pause',    "
                        + "      'video_hb','video_resume','video_stop')      "
                        + " )                                                 "
                        + "                                                   "
                        + " SELECT                                            "
                        + "   tmp.user_id,                                    "
                        + "   tmp.release_channel,                            "
                        + "   tmp.device_type,                                "
                        + "   tmp.event_id,                                   "
                        + "   tmp.play_id,                                    "
                        + "   tmp.video_id,                                   "
                        + "   tmp.event_time,                                 "
                        + "   dim.video_name,                                 "
                        + "   dim.video_type,                                 "
                        + "   dim.video_album,                                "
                        + "   dim.video_author,                               "
                        + "   dim.video_timelong,                             "
                        + "   dim.create_time,                                "
                        + "   -1 as start_time                                "
                        + " FROM tmp LEFT JOIN video_hbase                    "
                        + "   FOR SYSTEM_TIME AS OF tmp.pt AS dim             "
                        + "   ON LPAD(tmp.video_id,7,'0') = dim.video_id      "
        );


        /**
         * 将维度关联完成的表，转成流
         * 并通过流api，添加“start_time"字段
         */
        DataStream<Etl04Bean> ds = tenv.toDataStream(tenv.from("joined_view"), Etl04Bean.class);

        SingleOutputStreamOperator<Etl04Bean> resDs = ds.keyBy(bean -> bean.play_id)
                .process(new KeyedProcessFunction<String, Etl04Bean, Etl04Bean>() {
                    ValueState<Etl04Bean> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        state = getRuntimeContext().getState(new ValueStateDescriptor<Etl04Bean>("state", Etl04Bean.class));
                    }

                    @Override
                    public void processElement(Etl04Bean bean, KeyedProcessFunction<String, Etl04Bean, Etl04Bean>.Context ctx, Collector<Etl04Bean> out) throws Exception {

                        // 获取状态中的数据bean
                        Etl04Bean stateBean = state.value();

                        // 如果状态为空，说明是第一次接收到该播放的事件
                        // 则填充状态
                        if (stateBean == null) {
                            // 填充start_time
                            bean.setStart_time(bean.getEvent_time());
                            // 填充event_time
                            bean.setEvent_time(bean.getEvent_time());

                            // 放入状态
                            state.update(bean);
                            stateBean = state.value();
                        }

                        // 如果是resume事件，则更新状态中数据的start_time
                        if (bean.getEvent_id().equals("video_resume")) {
                            stateBean.setStart_time(bean.getEvent_time());
                        }

                        // 更新状态数据中的event_time 为本条时间的event_time,并输出
                        stateBean.setEvent_time(bean.getEvent_time());

                        out.collect(stateBean);
                    }
                });

        // 把结果流转成视图，以便后续可以用sql来做滚动时间窗口聚合
        tenv.createTemporaryView("tmp_view", resDs,
                Schema.newBuilder()
                        .column("user_id", DataTypes.BIGINT())
                        .column("release_channel", DataTypes.STRING())
                        .column("device_type", DataTypes.STRING())
                        .column("play_id", DataTypes.STRING())
                        .column("video_id", DataTypes.STRING())
                        .column("event_time", DataTypes.BIGINT())
                        .column("video_name", DataTypes.STRING())
                        .column("video_type", DataTypes.STRING())
                        .column("video_album", DataTypes.STRING())
                        .column("video_author", DataTypes.STRING())
                        .column("video_timelong", DataTypes.BIGINT())
                        .column("create_time", DataTypes.TIMESTAMP(3))
                        .column("start_time", DataTypes.BIGINT())
                        .columnByExpression("rt", "to_timestamp_ltz(event_time,3)")
                        .watermark("rt", "rt - interval '0' second ")
                        .build());

        // 创建doris连接器表，映射doris中的视频行为分析轻度聚合表
        tenv.executeSql(
                " create table video_doris (             "
                        + "     start_dt         DATE,             "
                        + "     user_id          BIGINT,           "
                        + "     release_channel  VARCHAR(20),      "
                        + "     device_type      VARCHAR(20),      "
                        + "     video_id         BIGINT ,          "
                        + "     video_play_id    VARCHAR(20),      "
                        + "     video_name       VARCHAR(20),      "
                        + "     video_type       VARCHAR(20),      "
                        + "     video_album      VARCHAR(40),      "
                        + "     video_author     VARCHAR(40),      "
                        + "     video_timelong   BIGINT,           "
                        + "     create_time      TIMESTAMP,        "
                        + "     play_start_time  BIGINT ,          "
                        + "     play_end_time    BIGINT            "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dws.video_events_agg01',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_tl" + System.currentTimeMillis() + "')"
        );


        // 滚动时间窗口分组聚合，求end_time最大值，并写入doris
        tenv.executeSql(
                " insert into video_doris  select                                  "
                        + "    to_date(date_format(window_start,'yyyy-MM-dd')) as start_dt, "
                        + "    user_id,                                                "
                        + "    release_channel,                                        "
                        + "    device_type,                                            "
                        + "    cast(video_id as bigint) as video_id,                   "
                        + "    play_id as video_play_id,                               "
                        + "    video_name,                                             "
                        + "    video_type,                                             "
                        + "    video_album,                                            "
                        + "    video_author,                                           "
                        + "    video_timelong,                                         "
                        + "    create_time,                                            "
                        + "    start_time as play_start_time,                          "
                        + "    max(event_time) as play_end_time                        "
                        + " from table(                                                "
                        + "  tumble(table tmp_view ,descriptor(rt),interval '5' minute)"
                        + " )                                                          "
                        + " GROUP BY                                                   "
                        + "    window_start,                                           "
                        + "    window_end,                                             "
                        + "    user_id,                                                "
                        + "    release_channel,                                        "
                        + "    device_type,                                            "
                        + "    play_id,                                                "
                        + "    video_id,                                               "
                        + "    video_name,                                             "
                        + "    video_type,                                             "
                        + "    video_album,                                            "
                        + "    video_author,                                           "
                        + "    video_timelong,                                         "
                        + "    create_time,                                            "
                        + "    start_time                                              "

        );


        env.execute();

    }
}
